package com.shujia.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/*
    写java连接并操作hbase程序
    实际上使用java操作的话，本质上是使用了JDBC驱动包的方式

    1、获取与数据库的连接对象
    2、获取操作数据的对象
    3、增删改查
    4、处理结果
    5、释放资源

    该程序要进行的操作
        1）创建一张表
        2）删除一张表
        3）向表中添加一列数据
        4）向表中批量添加数据
        5）查询一条数据
        6）查询批量数据
        7）结合过滤器查询
 */
public class HBaseApiDemo1 {

    private Connection conn;
    private Admin admin;

    /**
     * 获取与hbase数据连接对象
     */
    @Before
    public void getConnection() {
        try {
            //创建hbase配置文件对象
//        HBaseConfiguration conf = new HBaseConfiguration(); //旧版本的使用方式
            //HBaseConfiguration中有一个静态方法create() --> Configuration对象
            Configuration conf = HBaseConfiguration.create();
            //指定要连接hbase集群(zk集群)
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //获取数据库连接对象
            //使用连接工厂类创建连接对象
            conn = ConnectionFactory.createConnection(conf);

            //创建数据库操作对象
            //新版本已经没有HBaseAdmin这个类了
            admin = conn.getAdmin();

            System.out.println("成功获取数据库连接对象: " + conn);
            System.out.println("成功获取数据库操作对象: " + admin);
            System.out.println("====================================================");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 在hbase中创建一张表
     * <p>
     * create '表名','列簇1',...
     * 必须要有的元素是表名和列簇
     */
    @Test
    public void createOneTable() {
        try {
            //        HTableDescriptor students = new HTableDescriptor("students");
            // HTableDescriptor(final TableName name)在Hbase2.0之后就过时了
            //需要先将表名封装成TableName的对象
            //使用TableName中的静态方法public static TableName valueOf(String name)获取表名的对象
            TableName stu1 = TableName.valueOf("stu1");
            //创建表描述器
//        HTableDescriptor hTableDescriptor = new HTableDescriptor(stu1);
            //新版本希望使用 TableDescriptorBuilder 去创建表的对象
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(stu1);

            //向一个表中添加一个列簇
            //旧版本写法:addColumnFamily(final ColumnFamilyDescriptor family)
//        tableDescriptorBuilder.addColumnFamily()
            //新版本写法: setColumnFamily(final ColumnFamilyDescriptor family)

            //需要先将列簇封装成ColumnFamilyDescriptor(接口)对象
            // ColumnFamilyDescriptorBuilder
            ColumnFamilyDescriptor cf1 = ColumnFamilyDescriptorBuilder.of("cf1");

            //向一个表中添加一个列簇
            tableDescriptorBuilder.setColumnFamily(cf1);

            //调用方法创建该表
            //TableDescriptor
            admin.createTable(tableDescriptorBuilder.build());

            //hbase中数据,会被转化成特有类型的字节数组,无法使用java自带的工具类转字符串
            //使用Hbase自带的工具类转字符串Bytes(org.apache.hadoop.hbase.util)
            System.out.println(Bytes.toString(tableDescriptorBuilder.build().getTableName().getName()) + "表创建成功!!!");

        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 2）删除一张表
     * 先禁用再删除
     * disable 表名
     * drop 表名
     */
    @Test
    public void dropOneTable() {
        try {
            //将表名封装成TableName的对象
            TableName stu1 = TableName.valueOf("stu1");

            //先判断该表是否存在
            boolean b = admin.tableExists(stu1);
            //如果存在再禁用删除
            if (b) {
                //禁用表
                //void disableTable(TableName tableName)
                admin.disableTable(stu1);

                //删除表
                admin.deleteTable(stu1);

                System.out.println(Bytes.toString(stu1.getName()) + "表删除成功!!!");
            } else {
                System.out.println(Bytes.toString(stu1.getName()) + "表不存在!!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 3）向表中添加一列数据
     * put '表名','行键','列簇:列名','列值'   添加一列
     * put 'stu1','1001','cf1:name','雷鹏程'
     * <p>
     * rowkey的设计原则:
     * 1. 唯一性
     * 2. 长度不宜过长
     * 3. 散列性
     */
    @Test
    public void putOneData() {
        try {
            TableName tableName = TableName.valueOf("stu1");

            //表对象来调用方法添加数据
            //获取表对象
            Table stu1Table = conn.getTable(tableName);

            //将一列的数据封装成一个Put类的对象
            //Put(byte[] row) 传入行键
            Put put = new Put(Bytes.toBytes("1001"));
            //设置Put对象 第一种设置做法
//            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("name"), Bytes.toBytes("雷鹏程"));
            //设置Put对象 第二种设置做法
            //Put add(Cell cell)
            //KeyValue是Cell实现子类
            //KeyValue(final byte [] row, final byte [] family,final byte [] qualifier, final byte [] value)
            KeyValue keyValue = new KeyValue(Bytes.toBytes("1001"),
                    Bytes.toBytes("cf1"),
                    Bytes.toBytes("age"),
                    Bytes.toBytes(18));
            put.add(keyValue);

            //void put(Put put)
            stu1Table.put(put);
            System.out.println(Bytes.toString(tableName.getName()) + " 添加一列数据成功!!");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 向表中添加一行数据
     */
    @Test
    public void putMoreData() {
        try {
            TableName tableName = TableName.valueOf("students2");

            //表对象来调用方法添加数据
            //获取表对象
            Table stu1Table = conn.getTable(tableName);

            //创建List集合对象
            ArrayList<Put> puts = new ArrayList<>();
            //1500101001,张雷鹏程,23,男,张理科六班
            puts.add(new Put(Bytes.toBytes("1500101001")).addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("张雷鹏程")));
            puts.add(new Put(Bytes.toBytes("1500101001")).addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(23)));
            puts.add(new Put(Bytes.toBytes("1500101001")).addColumn(Bytes.toBytes("detail"), Bytes.toBytes("gender"), Bytes.toBytes("男")));
            puts.add(new Put(Bytes.toBytes("1500101001")).addColumn(Bytes.toBytes("detail"), Bytes.toBytes("clazz"), Bytes.toBytes("张理科六班")));

            //每一个列是由一个Put对象组成,那么,我们将一行中的所有列都封装成Put对象
            //将这么多的Put对象封装成List集合
            //void put(List<Put> puts)
            stu1Table.put(puts);

            System.out.println(Bytes.toString(tableName.getName()) + " 添加一行数据完毕!!");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 将本地的学生数据添加到hbase中
     * 可以使用代码循环的方式添加数据
     */
    @Test
    public void putFileData() {
        try {
            //创建表
            TableName tableName = TableName.valueOf("students2");

            boolean b = admin.tableExists(tableName);
            if (!b) {
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

                //添加列簇
                //创建ArrayList存储多个列簇
                ArrayList<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
                //创建两个列簇
                ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");
                ColumnFamilyDescriptor detail = ColumnFamilyDescriptorBuilder.of("detail");
                //将创建的列簇放入到集合中
                columnFamilyDescriptors.add(info);
                columnFamilyDescriptors.add(detail);

                //将列簇信息添加到表中
                tableDescriptorBuilder.setColumnFamilies(columnFamilyDescriptors);

                //创建该表
                admin.createTable(tableDescriptorBuilder.build());
            }


            //----------------------------------------------------------------------------------
            //获取表
            Table students = conn.getTable(tableName);

            //创建一个put集合对象
            ArrayList<Put> puts = new ArrayList<>();

            //创建字符缓冲输入流
            BufferedReader br = new BufferedReader(new FileReader("data/students.csv")); //在单元测试中路径以当前的项目为根目录

            String line = null;
            while ((line = br.readLine()) != null) {
                //1500100002,吕金鹏,24,男,文科六班
                String[] strings = line.split(",");
                /*
                    行键的设计原则：
                    1）唯一性
                    2）长度不宜过长
                    3）散列性

                    此刻案例中我们拿学生的学号作为行键进行存储
                 */
                byte[] id = Bytes.toBytes(strings[0]);
                //因为put在hbase中表示的是一列数据，所以要创建多个put对象，表示一行很多列
                puts.add(new Put(id).addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(strings[1])));
                puts.add(new Put(id).addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(strings[2])));
                puts.add(new Put(id).addColumn(Bytes.toBytes("detail"), Bytes.toBytes("gender"), Bytes.toBytes(strings[3])));
                puts.add(new Put(id).addColumn(Bytes.toBytes("detail"), Bytes.toBytes("clazz"), Bytes.toBytes(strings[4])));
            }

            students.put(puts);

            System.out.println("数据添加完毕！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 获取一条数据
     * <p>
     * get '表名','行键','列簇:列名'
     * <p>
     * 1500100008,符半双,22,女,理科六班
     * <p>
     * 数据根据三维度有序存储（行键，列簇，列名）默认都是按照字典顺序排序的
     */
    @Test
    public void getOneData() {
        try {
            //先将表名封装成TableName对象
            TableName tableName = TableName.valueOf("students2");

            //获取表的实例对象
            Table students = conn.getTable(tableName);
            //创建Get对象 传入行键
            Get get = new Get(Bytes.toBytes("1500100008"));

            //传入get对象
            Result result = students.get(get); //返回的是一行数据，有很多的列
//            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("姓名")));
//            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("年龄")));
//            String gender = Bytes.toString(result.getValue(Bytes.toBytes("detail"), Bytes.toBytes("性别")));
//            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("detail"), Bytes.toBytes("班级")));
//
//            System.out.println("姓名：" + name + ", 年龄：" + age + ", 性别：" + gender + ", 班级：" + clazz);

            StringBuilder sb = new StringBuilder();
            //获取行键
            String id = Bytes.toString(result.getRow());
            sb.append("学号：").append(id);

            //如果在不知道列名的情况下，就不能够使用上面的方式获取数据了。
            List<Cell> cells = result.listCells(); //获取多个单元格组成的List集合，每个Cell单元格中都包含了四个维度和一个值（行键，列簇，列名，版本号(默认最新)）--> 列值
            for (Cell cell : cells) {
                //使用Hbase自带的工具类CellUtil获取Cell单元格中的维度值
                String lieming = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                sb.append(",").append(lieming).append("：").append(value);
            }

            System.out.println(sb);


        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 获取一批数据
     * <p>
     * scan '表名'   查询所有数据
     */
    @Test
    public void scanData() {
        try {
            TableName tableName = TableName.valueOf("students2");

            //获取表的实例
            Table students = conn.getTable(tableName);

            //创建Scan对象
            Scan scan = new Scan(); //默认是从第一个列到最后一个列

            /*
                通常情况下，我们去查询的时候，并不会全部查询，因为没有意义，一般情况下，会进行开始行和结束行的设置[startRow, stopRow)
                本质上是对Scan做设置
             */
            //传入的是行键
            /**
             * 旧版本写法
             */
//            scan.setStartRow(Bytes.toBytes("1500100001"));
//            scan.setStopRow(Bytes.toBytes("1500100011"));
            /**
             * 新版本写法
             */
            scan.withStartRow(Bytes.toBytes("1500100001"));
            //Scan withStopRow(byte[] stopRow, boolean inclusive) inclusive参数表示包含该行
            scan.withStopRow(Bytes.toBytes("1500100011")); //如果不写表示不包含结束行


            //ResultScanner getScanner(Scan scan)
            //通过源码发现，ResultScanner中将每一行封装成Result类型的对象，将多个该行result对象放入到一个迭代器中
            ResultScanner resultScanner = students.getScanner(scan);
            //推荐使用增强for循环遍历，因为只有Collection集合中才有迭代器，又因为是Collection集合，所以可以采用增强for循环遍历
//            for (Result result : resultScanner) {
////                String id = Bytes.toString(result.getRow()); //获取行键
////                String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("姓名")));
////                String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("年龄")));
////                String gender = Bytes.toString(result.getValue(Bytes.toBytes("detail"), Bytes.toBytes("性别")));
////                String clazz = Bytes.toString(result.getValue(Bytes.toBytes("detail"), Bytes.toBytes("班级")));
////
////                System.out.println("学号：" + id + ", 姓名：" + name + ", 年龄：" + age + ", 性别：" + gender + ", 班级：" + clazz);
//                //调用方法打印
//                printResult(result);
//            }
            printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * hbase是主从结构，主节点中有HMaster 从节点：HRegionServer
     * RegionServer是负责管理region的，HMaster是负责分配region，让RegionServer相对负载均衡
     * <p>
     * hbase的热点问题：当某一个RegionServer上的一个Region会在同一时刻有大量的并发操作操作这个region，就会导致该RegionServer相比较于其他的RegionServer会比较繁忙
     * 有可能会导致该RegionServer挂掉。这就是热点问题。
     * <p>
     * 解决方法：预分region,提前设置好分割点，region的个数是分割点的数量+1，由Master自动的会进行负载均衡，最终会分摊一部分请求到其他的RegionServer上处理
     */
    @Test
    public void createPreviewTable() {
        try {
            TableName tableName = TableName.valueOf("pre_tb");

            //pre_tb
            //判断表是否存在
            boolean b = admin.tableExists(tableName);
            if (b) {
                //先禁用
                admin.disableTable(tableName);
                //删除表
                admin.deleteTable(tableName);
            }


            //创建分割键的二维数组
            byte[][] splits = {
                    Bytes.toBytes("e"),
                    Bytes.toBytes("l"),
                    Bytes.toBytes("s")
            };

            //创建代码
            //创建表描述器对象
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
            //创建列簇描述器
            //对于列簇而言，可以额外设置过期时间，版本号
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.newBuilder(ColumnFamilyDescriptorBuilder.of("info"))
//                    .setBloomFilterType()
                    .setTimeToLive(60 * 60 * 24) //设置过期时间 秒级别
                    .setMaxVersions(3) //设置最大版本号
                    .build();
            //将列簇添加到表中
            tableDescriptorBuilder.setColumnFamily(info);
            //创建表
//            admin.createTable(tableDescriptorBuilder.build());
            //createTable(TableDescriptor desc, byte[][] splitKeys) 创建region表
            admin.createTable(tableDescriptorBuilder.build(), splits); // e l s

            System.out.println("预分region表创建成功！！");


        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    /**
     * 打印批量查询的结果
     *
     * @param resultScanner 批量查询的结果
     */
    public void printResultScanner(ResultScanner resultScanner) {
        for (Result result : resultScanner) {
            printResult(result);
        }
    }

    /**
     * 打印一行中每一个列
     *
     * @param result 一行
     */
    public void printResult(Result result) {
        StringBuilder sb = new StringBuilder();
        String rowKey = Bytes.toString(result.getRow());
        sb.append("行键：").append(rowKey);
        List<Cell> cells = result.listCells(); //获取多个单元格组成的List集合，每个Cell单元格中都包含了四个维度和一个值（行键，列簇，列名，版本号(默认最新)）--> 列值
        for (Cell cell : cells) {
            //使用Hbase自带的工具类CellUtil获取Cell单元格中的维度值
            String lieming = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));

            sb.append(",").append(lieming).append("：").append(value);
        }
        System.out.println(sb);
    }


    /**
     * 用于释放资源
     */
    @After
    public void close() {
        try {
            if (conn != null) {
                conn.close();
            }

            if (admin != null) {
                admin.close();
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
